package com.alibaba.otter.node.etl.load.loader.mq.rabbit;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.api.ChannelManager;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.*;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.consume.*;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.exception.MqConfigException;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.model.ListenerProperties;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.provider.ConsumerTagProvider;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.provider.DefaultConsumerTagProvider;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.strategy.FailureAction;
import com.google.common.collect.Lists;
import com.rabbitmq.client.impl.recovery.AutorecoveringChannel;
import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
//
//@Slf4j
//class RabbitMqListener implements Listener {
//
//    @Setter
//    private String namespaceName;
//
//    @Setter
//    private String queue;
//
//    @Setter
//    private DisconnectedCallback disconnectedCallback = new DefaultDisconnectedCallback();
//
//    @Setter
//    private ConsumerTagProvider consumerTagProvider = new DefaultConsumerTagProvider();
//
//    @Setter
//    private int prefetchCount = 200;
//
//    @Setter
//    private ConnectedCallback connectedCallback = new DefaultConnectedCallback();
//
//    private List<ConsumeFailureCallback> failureCallbacks = Lists.newArrayList(new DefaultConsumeFailureCallback());
//
//    @Setter
//    private FailureAction failureAction;
//
//    private List<SuccessCallback> successCallbacks = Lists.newArrayList(new DefaultConsumeSuccessCallback());
//
//    @Setter
//    private int threadCnt = MqConstants.DEFAULT_CONSUME_THREAD_COUNT;
//
//    private DefaultHandlerInvoker handlerInvoker = new DefaultHandlerInvoker();
//
//    private BlockingQueueConsumer defaultConsumer;
//
//    private List<HeaderHandler> headerHandlers = Lists.newArrayList(
//            new RequestIdHandler()
//    );
//
//    private List<PropertyHandler> propertyHandlers = Lists.newArrayList();
//
//    public RabbitMqListener(String namespaceName) {
//        this.namespaceName = namespaceName;
//    }
//
//    public void addFailureCallbacks(List<ConsumeFailureCallback> failureCallbacks) {
//        this.failureCallbacks.clear();
//        this.failureCallbacks.addAll(failureCallbacks);
//    }
//
//    public void addSuccessCallbacks(List<SuccessCallback> successCallbacks) {
//        this.successCallbacks.addAll(successCallbacks);
//    }
//
//    public void setHandler(Handler handler) {
//        handlerInvoker.setHandler(handler);
//    }
//
//
//    public void addHeaderHandler(List<HeaderHandler> handlers) {
//        this.headerHandlers.addAll(handlers);
//    }
//
//    public void addPropertyHandler(List<PropertyHandler> handlers) {
//        this.propertyHandlers.addAll(handlers);
//    }
//
//    private ChannelManager channelManager = new ChannelManager();
//
//    @Override
//    public void listen() {
//        if (queue == null) {
//            String msg = "未指定待监听的queue";
//            log.error(msg, new MqConfigException(msg));
//            return;
//        }
//        if (defaultConsumer == null) {
//
//            AutorecoveringChannel channel = channelManager.getListenerChannel(namespaceName);
//            channelManager.setPrefetchCount(prefetchCount);
//            defaultConsumer = new BlockingQueueConsumer(channel, namespaceName);
//
//            ListenerProperties properties = new ListenerProperties();
//            properties.setNamespace(namespaceName);
//            properties.setQueue(queue);
//            properties.setPrefetchCnt(prefetchCount);
//            properties.setThreadCnt(threadCnt);
//
//            handlerInvoker.setHeaderHandlers(headerHandlers);
//            handlerInvoker.setPropertyHandlers(propertyHandlers);
//            handlerInvoker.setSuccessCallbacks(successCallbacks);
//
//            channel.addShutdownListener(e -> {
//                Thread t = new Thread(() -> {
//                    while (true) {
//                        try {
//                            AutorecoveringConnection conn = channelManager.getConn(namespaceName);
//                            log.info("recovering mq channel on connection: " + conn);
//                            channel.automaticallyRecover(conn, conn);
//                            break;
//                        } catch (Exception ex) {
//                            MqUtils.sleep(MqConstants.RECONNECT_INTERVAL_IN_MS);
//                        }
//                    }
//                });
//                t.start();
//            });
//
//            defaultConsumer.setHandlerInvoker(handlerInvoker);
//            defaultConsumer.setFailureAction(failureAction);
//            defaultConsumer.setThreadCount(threadCnt);
//            defaultConsumer.setDisconnectedCallback(disconnectedCallback);
//            defaultConsumer.setConnectedCallback(connectedCallback);
//            defaultConsumer.setFailureCallbacks(failureCallbacks);
//
//            while (true) {
//                try {
//                    String consumerTag = consumerTagProvider.provide(properties);
//                    defaultConsumer.setConsumerTag(consumerTag);
//                    channel.basicConsume(queue, false, consumerTag, defaultConsumer);
//                    break;
//                } catch (Exception e) {
//                    log.error(MqConstants.CONNECTION_ERROR_TIP + e.getMessage(), e);
//                    MqUtils.sleep(MqConstants.RECONNECT_INTERVAL_IN_MS);
//                }
//            }
//        }
//        defaultConsumer.listen(queue);
//    }
//
//    @Override
//    public void stop() {
//        defaultConsumer.stop();
//    }
//}
